package com.nulldev.util.concurrency.threadIt.v4.executors;

import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import com.nulldev.util.JVM.Arguments;
import com.nulldev.util.VariableAPI.ArrayUtils;
import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.utils.impls.TEConditions;
import com.nulldev.util.data.Arrays.arrays.FastArrayList;
import com.nulldev.util.data.FastMath.FastMath;
import com.nulldev.util.logging.Logger;
import com.nulldev.util.logging.LoggerFactory;

/**
 * The DATE (Dynamically Allocated Task Executor) executor. <br>
 * <br>
 * WARNING: This is an experimental executor. <br>
 * Designed to execute tasks on <code>n (default 16)</code> threads. <br>
 * The DATE executor will dynamically unallocate and allocate threads on demand.
 * (~300ms) <br>
 * <b>WARNING: This executor does not respect synchronisation.</b> <br>
 * <br>
 * There are 2 versions of DATE: DATE-Standard (current), DATE-Priority
 * (HIGH_PRIORITY, LOW_PRIORITY) <br>
 * <b>DATE-lightning brings over more faster queueing from the Lightning
 * executor.</b>
 * 
 * @author null
 * @since 3.0
 */
public class DATE extends AbstractExecutorService implements ExtendedExecutor {

	private static final com.nulldev.util.data.FastMath.Math Math = FastMath.get();
	public static final String NAME = "DATE-lightning";

	public static enum DATEThreadStatus {
		IDLE,
		RUNNING,
		STOPPED,
	}

	public abstract class DATEThread extends Thread implements Runnable {

		private final AtomicReference<DATEThreadStatus> status = new AtomicReference<DATEThreadStatus>(DATEThreadStatus.IDLE);
		private final AtomicBoolean localSigStop = new AtomicBoolean();

		public DATEThread() {
			super("DATE Thread");
		}

		public DATEThread(final int index) {
			super("DATEThread=" + index);
		}

		@Override
		public void run() {
			while (!sigStop() && !this.localSigStop.get()) {
				try {
					this.status.lazySet(DATEThreadStatus.RUNNING);
					final Runnable task = this.getTask();
					if (task == null) {
						this.status.set(DATEThreadStatus.IDLE);
						continue;
					}
					if (task instanceof Executable<?>)
						((Executable<?>) task).run(this);
					else
						task.run();
					this.status.lazySet(DATEThreadStatus.IDLE);
				} catch (Throwable e) {
					log.error(e);
					this.status.lazySet(DATEThreadStatus.IDLE);
					try {
						Thread.sleep(DATE.this.getIdleTime());
					} catch (InterruptedException e1) {
					}
				}
			}
		}

		public DATEThreadStatus getStatus() {
			return this.status.get();
		}

		public DATEThreadStatus status() {
			return this.status.get();
		}

		public abstract boolean sigStop();

		public abstract boolean hasTasks();

		public abstract Runnable getTask();

		public boolean stopThread() {
			if (this.isActive())
				return false;
			if (this.isDaemon()) {
				this.setDaemon(false);
			}
			this.localSigStop.set(true);
			while (!this.isInterrupted())
				this.interrupt();
			this.status.lazySet(DATEThreadStatus.STOPPED);
			return true;
		}

		public boolean isActive() {
			return this.status.get() == DATEThreadStatus.RUNNING;
		}

		public boolean isIdle() {
			return this.status.get() == DATEThreadStatus.IDLE;
		}
	}

	private final LinkedTransferQueue<Runnable> tasks = new LinkedTransferQueue<Runnable>();
	private final List<DATEThread> executors = new FastArrayList<DATEThread>(true);
	private final static Logger log = LoggerFactory.getLogger();

	/*
	 * Reserve executor for self-management.
	 */
	private final ExecutorService core_date = Executors.newFixedThreadPool(2, new ThreadFactory() {

		@Override
		public Thread newThread(Runnable r) {
			return new Thread(r, "DATE: Managment Thread");
		}
	});

	public static final int DEFAULT_THREAD_COUNT = 16;
	public static final long SLEEP_BETWEEN_COMMANDS_DEFAULT = 75L; // min: 1, max: n
	public volatile long SLEEP_BETWEEN_COMMANDS = SLEEP_BETWEEN_COMMANDS_DEFAULT;
	public volatile boolean HIGH_PRIORITY_THREADING = Arguments.hasArgument("--DATE.useHPT");
	private final AtomicBoolean __sig_stop_global = new AtomicBoolean();
	private final AtomicBoolean __sig_stop_finished = new AtomicBoolean();

	private long getIdleTime() {
		final int tasks = this.tasks.size();
		if (tasks == 0) {
			return 255;
		} else if (tasks <= 1) {
			return 190;
		} else if (tasks <= 6) {
			return 135;
		} else {
			return this.SLEEP_BETWEEN_COMMANDS;
		}
	}

	private final int threadLimit;

	public DATE() {
		this(DEFAULT_THREAD_COUNT);
	}

	public DATE(final int threadCount) {
		this(threadCount, false);
	}

	public DATE(final int threadCount, final boolean high_priority) {
		this.threadLimit = threadCount;
		this.__init_threads(high_priority);
	}

	private void __init_threads(final boolean high_priority) {
		if (high_priority) {
			log.warn("__init_threads(bool) -> High priority execution enabled! This may cause system hangups!");
		}
		for (int i = 0; i < threadLimit; i++) {
			final DATEThread dt = new DATEThread(i) {

				@Override
				public boolean sigStop() {
					return DATE.this.__sig_stop_global.get();
				}

				@Override
				public boolean hasTasks() {
					return !DATE.this.tasks.isEmpty();
				}

				@Override
				public Runnable getTask() {
					try {
						return DATE.this.tasks.poll(5, TimeUnit.SECONDS);
					} catch (Exception e) {
						return null;
					}
				}
			};
			if (high_priority) {
				dt.setPriority(Thread.MAX_PRIORITY);
			}
			dt.start();
			this.executors.add(dt);
		}
	}

	private void run_task(final Executable<?> task) {
		TEConditions.validExecutable(task);
		this.__dynamically_reallocate(this.HIGH_PRIORITY_THREADING);
		try {
			if (task instanceof Runnable) {
				if (!this.tasks.tryTransfer((Runnable) task, 300L, TimeUnit.MILLISECONDS))
					this.tasks.put((Runnable) task);
			} else {
				final Runnable ex = () -> {
					task.run_st();
				};
				if (!this.tasks.tryTransfer(ex, 300L, TimeUnit.MILLISECONDS))
					this.tasks.put(ex);
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	private void __dynamically_reallocate(final boolean high_priority) {
		if (this.executors.size() == 1 || this.executors.size() < threadLimit) {
			for (int i = this.executors.size(); i < threadLimit; i++) {
				final DATEThread dt = new DATEThread(i) {

					@Override
					public boolean sigStop() {
						return DATE.this.__sig_stop_global.get();
					}

					@Override
					public boolean hasTasks() {
						return !DATE.this.tasks.isEmpty();
					}

					@Override
					public Runnable getTask() {
						try {
							return DATE.this.tasks.poll(5, TimeUnit.SECONDS);
						} catch (Exception e) {
							return null;
						}
					}
				};
				if (this.HIGH_PRIORITY_THREADING) {
					dt.setPriority(Thread.MAX_PRIORITY);
				}
				dt.start();
				this.executors.add(dt);
			}
		} else if (this.executors.size() >= threadLimit) {
			final int threadsToKill = MathUtil.clamp(3, 8, threadLimit / 4);
			int threadsKilled = 0;
			if (threadsToKill >= this.executors.size())
				return;
			for (int i = this.executors.size() - 1; i > 3; i--) {
				if (threadsKilled >= threadsToKill)
					break;
				try {
					final DATEThread dt = this.executors.get(i);
					if (dt.stopThread()) {
						this.executors.remove(i);
						threadsKilled++;
					}
				} catch (IndexOutOfBoundsException ex) {
					continue;
				}
			}
		}
	}

	@Override
	public void execute(final Runnable r) {
		TEConditions.validRunnable(r);
		DATE.this.__dynamically_reallocate(DATE.this.HIGH_PRIORITY_THREADING);
		try {
			if (!this.tasks.tryTransfer(r, 300, TimeUnit.MILLISECONDS))
				this.tasks.put(r);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	private void __sig_stop_idle() {
		if (this.executors.size() <= 1) {
			return;
		}
		for (final DATEThread dt : this.executors) {
			if (!dt.isActive()) {
				dt.stopThread();
				this.executors.remove(dt);
			}
		}
	}

	private void __sig_term() {
		this.__sig_stop_global.lazySet(true);
		for (final DATEThread dt : this.executors) {
			dt.stopThread();
			this.executors.remove(dt);
		}
		this.__sig_stop_finished.lazySet(true);
		this.core_date.shutdownNow();
	}

	@Override
	public boolean awaitTermination(final long amount, final TimeUnit unit) throws InterruptedException {
		this.core_date.execute(new Runnable() {

			@Override
			public void run() {
				DATE.this.__sig_term();
			}
		});
		Thread.sleep(unit.toMillis(amount));
		return this.isTerminated();
	}

	public void addTask(final Executable<?> task) {
		this.run_task(task);
	}

	@Override
	public void executor_submitAll(final List<Runnable> r) {
		if (r == null || r.isEmpty())
			return;
		DATE.this.__dynamically_reallocate(DATE.this.HIGH_PRIORITY_THREADING);
		try {
			Runnable t;
			while ((t = r.remove(0)) != null && !r.isEmpty()) {
				this.tasks.offer(t, 300, TimeUnit.MILLISECONDS);
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void executor_drainTasks(final Queue<Runnable> async_queue, final int batchSize) {
		if (async_queue == null || async_queue.isEmpty())
			return;
		if (batchSize <= 0)
			return;
		DATE.this.__dynamically_reallocate(DATE.this.HIGH_PRIORITY_THREADING);
		for (int i = 0; i < batchSize; i++) {
			Runnable r = null;
			try {
				r = async_queue.poll();
			} catch (Exception e) {
				continue;
			} finally {
				if (r == null)
					continue;
				try {
					if (!this.tasks.tryTransfer(r, 233L, TimeUnit.MILLISECONDS))
						this.tasks.put(r);
				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}
		}
	}

	@Override
	public boolean isShutdown() {
		return this.__sig_stop_global.get();
	}

	@Override
	public boolean isTerminated() {
		return this.__sig_stop_finished.get();
	}

	@Override
	public void shutdown() {
		// Shuts down threads after their tasks finish.
		this.__sig_stop_global.lazySet(true);
	}

	/**
	 * Shuts down any inactive threads. <br>
	 * These threads will be reallocated when needed.
	 */
	public void shutdownIdle() {
		this.__sig_stop_idle();
	}

	@Override
	public List<Runnable> shutdownNow() {
		final BlockingQueue<Runnable> tasks = this.tasks; // .clone();
		// this.tasks.clear();
		this.__sig_term();
		final List<Runnable> converted = new ArrayList<Runnable>(tasks.size());
		for (final Runnable t : tasks) {
			converted.add(t);
		}
		return converted;
	}

	public int getActiveThreads() {
		int i = 0;
		for (int x = 0; x < this.executors.size(); x++) {
			if (this.executors.get(x).isActive()) {
				i += 1;
			}
		}
		return i;
	}

	@Override
	public int executor_getFreeSlots() {
		int idleThreads = 0;
		for (final DATEThread dt : this.executors)
			if (dt.isIdle())
				idleThreads += 1;
		return MathUtil.clamp(1, this.threadLimit, idleThreads);
	}

	@Override
	public boolean executorSupportsSlotCount() {
		return true;
	}

	private static final ExecutorFeatureExtensions[] SUPPORTED = new ExecutorFeatureExtensions[]
		{ ExecutorFeatureExtensions.EEX_core_executor_slots, ExecutorFeatureExtensions.EEX_query_slots_supported,
				ExecutorFeatureExtensions.EEX_core_submit_all_tasks, ExecutorFeatureExtensions.EEX_core_drain_task_queue,
				ExecutorFeatureExtensions.EEX_executor_await_idle, ExecutorFeatureExtensions.EEX_query_batching_supported,
				ExecutorFeatureExtensions.EEX_query_slots_supported, ExecutorFeatureExtensions.EEX_query_thread_affinity_supported,
				ExecutorFeatureExtensions.EEX_executor_native_batching, ExecutorFeatureExtensions.EEX_executor_attemptable_submissions,
				ExecutorFeatureExtensions.EEX_executor_attemptable_submissions2, ExecutorFeatureExtensions.EEX_executor_attemptable_batch_submissions, };

	@Override
	public ExecutorFeatureExtensions[] supported() {
		return SUPPORTED;
	}

	@Override
	public boolean supportsFeature(final ExecutorFeatureExtensions feature) {
		return ArrayUtils.contains(SUPPORTED, feature);
	}

	@Override
	public String toString() {
		return "DATE-lighting[type=STANDARD,threads=" + this.threadLimit + ",active=" + this.getActiveThreads() + "]";
	}

	@SuppressWarnings("static-access")
	@Override
	public void awaitIdle() {
		final int limit = 8;
		int waitingInt = limit;
		final int s = (int) (35 / limit);
		while (!this.tasks.isEmpty()) {
			try {
				if (waitingInt <= 0) {
					waitingInt = limit + 1;
				}
				Thread.sleep((long) Math.ceil(s / waitingInt));
				waitingInt -= 1;
			} catch (InterruptedException e) {
			}
		}
	}

	/* EEX_query_batching_supported */

	@Override
	public boolean executorSupportsBatching() {
		return true;
	}

	/* EEX_query_thread_affinity_supported */

	@Override
	public boolean executorSupportsThreadAffinity() {
		return false;
	}

	@Override
	public boolean executorSupportsThreadPinning() {
		return false;
	}

	/* EEX_executor_attemptable_submissions */

	@Override
	public boolean supportsAttemptableSubmissions() {
		return true;
	}

	@Override
	public boolean attempt_submit(final Runnable command) {
		if (command == null)
			return false;
		try {
			return this.tasks.offer(command, 233L, TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			return false;
		}
	}

	/* EEX_executor_attemptable_submissions2 */

	@Override
	public boolean supportsAttemptableSubmissions2() {
		return true;
	}

	@Override
	public <T> boolean attempt_submit2(final Executable<T> task) {
		if (task == null)
			return false;

		if (this.tasks.offer((Runnable) task, 233L, TimeUnit.MILLISECONDS)) {
			task.signalQueued();
			return true;
		} else {
			task.rejectSubmission();
			return false;
		}
	}

	/* EEX_executor_attemptable_batch_submissions */

	@Override
	public boolean supportsAttemptableBatchSubmissions() {
		return true;
	}

	@Override
	public boolean attempt_batch(final Runnable command) {
		if (command == null)
			return false;
		try {
			return this.tasks.offer(command, 233L, TimeUnit.MILLISECONDS);
		} catch (Exception e) {
			return false;
		}
	}

	/* EEX_executor_native_batching */

	@Override
	public void batch(final Runnable f) {
		TEConditions.validRunnable(f);
		try {
			if (!this.tasks.tryTransfer(f, 233L, TimeUnit.MILLISECONDS))
				this.tasks.put(f);
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public void batchAll(final Runnable[] _r) {
		if (_r == null || _r.length == 0)
			return;
		for (final Runnable f : _r) {
			try {
				if (!this.tasks.tryTransfer(f, 233L, TimeUnit.MILLISECONDS))
					this.tasks.put(f);
			} catch (Exception e) {
				throw new RuntimeException(e);
			}
		}
	}

	/* EEX_query_debug */

	@Override
	public Object debug_query(final EEXDebugQueryType type) {
		switch (type) {
			case EEX_DEBUG_TOTAL_THREAD_COUNT:
				return this.threadLimit;
			case EEX_DEBUG_ACTIVE_THREAD_COUNT:
				return this.getActiveThreads();
			case EEX_DEBUG_TASKS_IN_QUEUE:
				return this.tasks.size();
			/* */
			default:
				return null;
		}
	}
}
